package com.atguigu.dwm.lineage.service.impl;

import com.alibaba.fastjson.JSON;
import com.atguigu.dwm.common.util.SourceTableDispatcher;
import com.atguigu.dwm.common.util.SqlParser;
import com.atguigu.dwm.ds.bean.TDsTaskDefinition;
import com.atguigu.dwm.ds.service.TDsTaskDefinitionService;
import com.atguigu.dwm.lineage.bean.LineageInfo;
import com.atguigu.dwm.lineage.mapper.LineageInfoMapper;
import com.atguigu.dwm.lineage.service.LineageInfoService;
import com.atguigu.dwm.model.bean.DmTable;
import com.atguigu.dwm.model.service.DmTableService;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.baomidou.dynamic.datasource.annotation.DS;

import java.util.*;

/**
 * <p>
 * 血缘关系表 服务实现类
 * </p>
 *
 * @author zhangchen
 * @since 2024-07-08
 */
@Service
@DS("dwm")
public class LineageInfoServiceImpl extends ServiceImpl<LineageInfoMapper, LineageInfo> implements LineageInfoService {


    @Autowired
    DmTableService dmTableService;

    @Autowired
    TDsTaskDefinitionService tDsTaskDefinitionService;

    //1   提取有目前所有表的元数据
    //2   提取ds（调度脚本的服务）的sql定义    定时调用后台方法
    //3   分析sql 提取 表的上下游关系  定时调用后台方法
    //4   保存上下游关系    定时调用后台方法
    public void extractLineageInfo() throws Exception {
        //1   提取有目前所有表的元数据
        List<DmTable> dmTableList = dmTableService.list(new QueryWrapper<DmTable>().eq("is_deleted", 0));
        //2  提取ds（调度脚本的服务）的sql定义    定时调用后台方法
        List<TDsTaskDefinition> taskDefinitionList = tDsTaskDefinitionService.getTaskDefinitionList(dmTableList);
        System.out.println("taskDefinitionList = " + taskDefinitionList);
        //3 为了方便整理上下游关系准备两个HashMap
          Map<String, Set<String>> sourceMap = new HashMap<>(taskDefinitionList.size());
          Map<String, Set<String>> sinkMap = new HashMap<>(taskDefinitionList.size());
          for (TDsTaskDefinition tDsTaskDefinition : taskDefinitionList) {
              String sql = tDsTaskDefinition.getSql();

              SourceTableDispatcher dispatcher = new SourceTableDispatcher();
              //把输出表的库名作为输入表的默认名称 ： 基本假设
              String curFullname = tDsTaskDefinition.getName();
              if(curFullname.indexOf(".")>0){
                  String schemaName = curFullname.substring(0, curFullname.indexOf("."));
                  dispatcher.setDefaultSchemaName(schemaName);
              }else{
                  continue;
              }
              SqlParser.parse(sql,dispatcher);

              Set<String> sourceTableSet = dispatcher.getSourceTableSet();
              Set<String>  subqueryTableSet = dispatcher.getSubqueryTableSet();
              sourceTableSet.removeAll(subqueryTableSet);//排除子查询
              sourceTableSet.remove(curFullname);//排除自己
              //装配进 sourcehashMap中
              sourceMap.put(curFullname,sourceTableSet);
              //装配进 sinkMap中 key 来源表  value 是 输出表的集合
              for (String sourceTablename : sourceTableSet) {
                  Set<String> sinkTableSet = sinkMap.get(sourceTablename);
                  if(sinkTableSet==null){
                      sinkTableSet = new HashSet<String>();
                  }
                  sinkTableSet.add(curFullname);
                  sinkMap.put(sourceTablename,sinkTableSet);
              }



          }

        //4   保存上下游关系 到lineage_info
        List<LineageInfo> lineageInfoList=new ArrayList<>(dmTableList.size());
        for (DmTable dmTable : dmTableList) {
            LineageInfo lineageInfo = new LineageInfo();
            lineageInfo.setTableName(dmTable.getTableName());
            lineageInfo.setSchemaName(dmTable.getSchemaName());

            Set<String> sourceTableSet = sourceMap.get(dmTable.getSchemaName() + "." + dmTable.getTableName());
            if(sourceTableSet!=null&&sourceTableSet.size()>0){
                lineageInfo.setSourceTables(StringUtils.join(sourceTableSet,","));
            }
            Set<String> sinkTableSet = sinkMap.get(dmTable.getSchemaName() + "." + dmTable.getTableName());

            if(sinkTableSet!=null&&sinkTableSet.size()>0){
                lineageInfo.setSinkTables(StringUtils.join(sinkTableSet,","));
            }
            lineageInfo.setLastUpdateTime(new Date());
            lineageInfoList.add(lineageInfo);
        }
        this.remove(new QueryWrapper<LineageInfo>());
        this.saveOrUpdateBatch(lineageInfoList);


    }



}
